use std::borrow::BorrowMut;
use std::cell::RefCell;
use std::collections::HashMap;
use std::future::Future;
use std::iter::Map;
use std::ops::{Deref, DerefMut};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use serde_json::json;
use tokio::fs::File;
use tokio::io::{AsyncWriteExt, AsyncReadExt};
use tokio::sync::oneshot::channel;
use tokio::time::Instant;
use tokio_stream::Stream;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Code, Request, Response, Status, Streaming};
use crate::protocol_buffer::exts::{ToJson, FromJson};
use crate::protocol_buffer::gen::{CreatePipeRequest, CreatePipeResponse, GenericResult, ListPipesAndAppsResponse, PipeConnectionRequest, PipeConnectionResponse, PublishPipeMessageRequest, PublishPipeMessageResponse, RetrieveMessageRequest, RetrieveMessageResponse, UpdatePipeResponse, UpdatePipeRequest};
/**
 * @author Almas software technologies muhtarjaan mahmood (مۇختەرجان مەخمۇت)
 * @email ug-project@outlook.com
 * @create date 2022-07-07 11:53:57
 * @modify date 2022-07-07 11:53:57
 * @desc [description]
*/

use crate::protocol_buffer::gen::legan_piping_server::LeganPiping;
use crate::{MessagePipe, SingleFileStorageSystem, StorageSystem, StoringImportance, protocol_buffer};
use crate::legan_piping::message::Message;
use crate::logger::logger::{Logger, TerminalLogger};
use crate::protocol_buffer::entity::{Empty, TransformerApplication};
use super::exts::Apply;
use super::transformer_application_extends::TransformerApplicationExtends;
pub struct LeganPipingServerImpl<LOGGER: Logger + Send + Sync + 'static> {
    storage_systems: Vec<Arc<dyn StorageSystem + Send + Sync>>,
    pub pipes: Arc<Mutex<Vec<MessagePipe<LOGGER>>>>, //Why Arc? Because we need to access it in multiple threads/tasks.
    pub apps: Mutex<Vec<TransformerApplicationExtends>>,
    logger: Arc<LOGGER>,
}

impl<LOGGER: Logger + Send + Sync + 'static> LeganPipingServerImpl<LOGGER> {
    pub async fn new(logger: Arc<LOGGER>) -> LeganPipingServerImpl<LOGGER> {
        let default_storage = Arc::new(SingleFileStorageSystem::new_with_file_name("data.bin".into(), 3));
        let mut pipes = vec![];
        let mut apps = vec![];
        if let Ok(mut config_file) = File::open("Server.json").await { //If file is exists
            let mut json_string = "".into();
            if let Ok(size) = config_file.read_to_string(&mut json_string).await { //Read into string
                let result: serde_json::Value = serde_json::from_str(json_string.as_str()).expect("Cannot parse json"); //And parse
                result["apps"].as_array().expect("Json error").iter().for_each(|item|{ //And read 'apps' property into apps.
                    apps.push(TransformerApplication::from_json(item).into());
                });
                result["pipes"].as_array().expect("Json error").iter().for_each(|item|{ //And read 'pipes' property into apps.
                    pipes.push(MessagePipe::from_json(default_storage.clone(), item, logger.clone()));
                });
                logger.ok("Server config data is restored.");
            }
        } else {
            logger.info("There is no server config file so skip from read and parse");
        }
        let server = LeganPipingServerImpl {
            pipes: Arc::new(Mutex::new(pipes)),
            apps: Mutex::new(apps),
            logger,
            storage_systems: vec![
                default_storage
            ]
        };
        server.logger.ok("Start of server.");
        server
    }
    ///Save pipes and apps existing in this server.
    pub async fn save_pipes_and_apps(&self) {
        let mut file = tokio::fs::OpenOptions::new()
            .write(true)
            .read(false)
            .create(true)
            .append(false)
            .truncate(true)
            .open("Server.json").await.expect("Config File cannot be created or opened.");
        let apps = {
            Vec::from_iter(self.apps.lock().expect("Cannot acquire 'apps' lock").iter().map(|app| app.app.to_json()))
        };
        let pipes = {
            Vec::from_iter(self.pipes.lock().expect("Cannot acquire 'pipe' lock").iter().map(|pipe| pipe.to_json()))
        };
        let json = json!({
            "apps": apps,
            "pipes": pipes
        });
        let json = json.to_string();
        file.write_all(json.as_bytes()).await.expect("Cannot write server config file");
        file.flush().await.expect("Cannot flush server config file");
        self.logger.ok("Save apps and pipes");
    }
    fn check_pipe_is_exists(&self, pipe_uuid: &str) -> bool {
        let exists = {
            {
                let pipes = self.pipes.lock().unwrap();
                pipes.iter().any(|item| item.uuid == pipe_uuid)
            }
        };
        exists
    }
    fn check_app_is_exists(&self, app_uuid: &str) -> bool {
        if let Some(_) = self.apps.lock().expect("Cannot acquire apps lock").iter().filter(|item| item.app.uuid == app_uuid).next() {
            true
        } else {false}
    }
}

#[async_trait::async_trait]
impl<LOGGER: Logger + Send + Sync + 'static> LeganPiping for LeganPipingServerImpl<LOGGER> {
    async fn create_pipe(&self, request: Request<CreatePipeRequest>) -> Result<Response<CreatePipeResponse>, Status> {
        let _request_data = request.into_inner();
        let pipe = _request_data.pipe.ok_or(Status::new(Code::InvalidArgument, "Pipe data cannot be empty"))?;

        if self.check_pipe_is_exists(pipe.uuid.as_str()) {
            self.logger.error(format!("Create of '{}' because it is already exists.", pipe.name));
            return Err(Status::new(Code::AlreadyExists, "The pipe is already exists so this operation is being to cancelled."));
        }
        {
            let value = MessagePipe::new(
                pipe.uuid.clone(),
                pipe.name.clone(),
                pipe.description.clone(),
                match pipe.storing_importance {
                    0 => StoringImportance::Critical,
                    1 => StoringImportance::Usual,
                    _ => StoringImportance::Usual,
                },
                pipe.has_table,
                self.storage_systems.first().unwrap().clone(),
                self.logger.clone(),
            );
            { //Because the shorter scope hold the lock shorter time as possible
                self.pipes.lock().unwrap().deref_mut().push(value);
            }
        }
        self.logger.ok(format!("Create of '{}'", pipe.name));
        self.save_pipes_and_apps().await;
        Ok(Response::new(CreatePipeResponse{
            pipe: Some(pipe.clone()),
            result: Some(GenericResult{message: "Succeed.".into(), succeed: true})
        }))
    }

    async fn update_pipe(&self, request: Request<UpdatePipeRequest>) -> Result<Response<UpdatePipeResponse>, Status> {
        let request = request.into_inner();
        let pipe = request.pipe.expect("Pipe is not found in request.");
        if !self.check_pipe_is_exists(pipe.uuid.as_str()) {
            self.logger.error(format!("Create of '{}' because it is not exists.", pipe.name));
            return Err(Status::new(Code::AlreadyExists, "The pipe is not exists so this operation is being to cancelled."));
        }
        {
            let mut guard = self.pipes.lock().expect("Cannot acquire pipe lock");
            let vec = guard.borrow_mut();
            vec.iter_mut().filter(|item| item.uuid == pipe.uuid).take(1).next().expect("Cannot find pipe").apply(|me|{
                me.name = pipe.name.clone();
                me.description = pipe.description.clone();
                me.has_table = pipe.has_table;
                me.message_importance = match pipe.storing_importance {
                    0 => StoringImportance::Critical,
                    1 => StoringImportance::Usual,
                    _ => StoringImportance::Usual,
                };
            });
        }
        self.save_pipes_and_apps().await;
        Ok(Response::new(UpdatePipeResponse {
            pipe: Some(pipe.clone()),
            result: Some(GenericResult{
                message: "Succeed.".into(),
                succeed: true,
            })
        }))
    }

    async fn connect_to_pipe(&self, request: Request<PipeConnectionRequest>) -> Result<Response<PipeConnectionResponse>, Status> {
        let request = request.into_inner();
        let mut app: TransformerApplicationExtends = request.application_information.expect("App is not empty").into();
        // if !self.check_app_is_exists(app.app.uuid.as_str()) { //We shouldn't do this because the app is not exists at the first time connection.
        //     return Result::Err(Status::already_exists("the app is already exists in app list."));
        // }
        { //Check app input and output ports are used same pipe or not.
            let mut temp = HashMap::new();
            app.app.input_port.iter().for_each(|item|{
                temp.insert(item.pipe_id.clone(), item.pipe_id.clone());
            });
            let wrong_ports = app.app.output_port.iter().filter(|item|temp.contains_key(&item.pipe_id));
            if wrong_ports.clone().count() != 0 {
                let ports_list = wrong_ports.map(|item| format!("{}({})", &item.name, &item.pipe_id)).reduce(|a, b| format!("{}, {}", a, b)).expect("You should't be see this message, if you can see this error, this is represents a software bug.");
                self.logger.error(format!("App '{}' is Attempting to connect to server with non exists ports: [{}]", app.app.name, ports_list));
                return Result::Err(Status::invalid_argument(format!("The following ports are used both as input and output in your app, the ports are: {}", ports_list)));
            }
        };
        let (all_pipe_inputs, all_pipe_names) = {
            let mut outputs = HashMap::new();
            let mut inputs = HashMap::new();
            let guard = self.pipes.lock().expect("Cannot acquire the lock");
            guard.deref().iter().for_each(|item|{
                outputs.insert(item.uuid.clone(), item.input.clone());
            });
            guard.deref().iter().for_each(|item|{
                inputs.insert(item.uuid.clone(), item.uuid.clone());
            });
            (outputs, inputs)
        };
        {
            let not_found_input_ports = app.app.input_port.iter().filter(|port| !all_pipe_names.contains_key(&port.pipe_id)); //Collect all not founded input ports
            let not_found_output_ports = app.app.output_port.iter().filter(|port| !all_pipe_inputs.contains_key(&port.pipe_id)); //And collect all not founded output ports too.
            let all_not_founds = not_found_input_ports.clone().chain(not_found_output_ports.clone()); //Combine them into a single iterator
            if all_not_founds.clone().count() != 0 {
                //Make a list of all not founded ports list for better error information.
                let lists = all_not_founds.clone().map(|item| format!("{}({})", item.name.clone(), item.pipe_id)).reduce(|a, b| format!("{}, {}", a, b)).expect("This should be none, I'm so sorry for you see this, this is a software bug.");
                return Result::Err(Status::invalid_argument(format!("The following input/output port that required this application is not found in this server, there are the not founded input/output ports: [{}]", lists)));
            }
        }
        {
            app.app.output_port.iter().map(|item| all_pipe_inputs.get(&item.pipe_id).expect("Cannot be null")).for_each(|pipe_input|{
                app.output.push(pipe_input.clone().unwrap());
            });
        }
        {
            let mut guard = self.apps.lock().expect("Cannot acquire apps lock");
            {//Remove if exists.
                let mut index = -1i32;
                for item in guard.iter() {
                    if item.app.uuid == app.app.uuid {
                        break;
                    }
                    index += 1;
                }
                if index != -1 {
                    guard.remove(index as usize);
                    self.logger.ok(format!("Deletion of pre existing app: '{}'", app.app.name));
                }
            }
            self.logger.ok(format!("Connection of app: '{}'", app.app.name));
            guard.push(app);
        }
        Result::Ok(Response::new(PipeConnectionResponse { succeed: true, result: Some(GenericResult::succeed()) }))
    }

    type retrievePipeMessageStream = Pin<Box<dyn Stream<Item = Result<RetrieveMessageResponse, tonic::Status>> + Send>>;

    async fn retrieve_pipe_message(&self, request: Request<Streaming<RetrieveMessageRequest>>) -> Result<Response<Self::retrievePipeMessageStream>, Status> {
        let mut request_stream = request.into_inner();
        let app_uuid = request_stream.message().await.expect("Cannot get app id").expect("app id is cannot be null").app_uuid;
        if !self.check_app_is_exists(app_uuid.as_str()) {
            self.logger.error(format!("Of Retrieve pipe message of app: 'uuid: {}'", app_uuid));
            return Result::Err(Status::not_found("The app that you provided is not found, please check your app id in your request."));
        }
        let (result_sender, result_receiver) = tokio::sync::mpsc::channel::<Result<RetrieveMessageResponse, Status>>(128);
        let (message_sender, mut message_receiver) = tokio::sync::mpsc::channel(1024);
        {
            self.apps.lock().expect("Cannot acquire apps lock").deref_mut().iter_mut().filter(|app| app.app.uuid == app_uuid).next().expect("App is should be not empty").input = Some(message_sender.clone());
        }
        {
            let mut guard = self.pipes.lock().expect("Cannot acquire pipes lock");
            guard.deref_mut().iter_mut().filter(|pipe| pipe.uuid == app_uuid).for_each(|pipe|{
                pipe.output.lock().expect("Lock is required while retrieve_pipe_message").insert(app_uuid.clone(), message_sender.clone());
            });
        }
        let (pipes_mutex, _app_uuid, ) = (self.pipes.clone(), app_uuid.clone());
        tokio::spawn(async move {
            while let Some(message) = message_receiver.recv().await {
                result_sender.send(Result::Ok(RetrieveMessageResponse {
                    messages: vec![message.into()],

                })).await.expect("Result is already closed.");
            }
            { //Clear disconnected connection
                let mut guard = pipes_mutex.lock().expect("Cannot acquire pipes lock");
                guard.iter_mut().for_each(|pipe|{
                    pipe.output.lock().expect("Acquire lock failed while un-register port").remove(&_app_uuid);
                });
            }
        });
        let stream = ReceiverStream::new(result_receiver);
        self.logger.ok(format!("Retrieving pipe message of app: 'uuid: {}'", app_uuid));
        Result::Ok(Response::new(Box::pin(stream)))
    }

    async fn publish_pipe_message(&self, request: Request<PublishPipeMessageRequest>) -> Result<Response<PublishPipeMessageResponse>, Status> {
        let all_message = request.get_ref().message.iter().map(|message| message.clone().into());
        let app_id = request.get_ref().app_uuid.as_str();
        if !self.check_app_is_exists(app_id) {
            return Result::Err(Status::not_found("The app that you provided is not found, please check your app id in your request."));
        }
        let senders = { //Actually the senders is input ports of message pipes so write to these pipe we can directly write into message pipe.
            let guard = self.apps.lock().expect("Cannot acquire apps lock");
            Vec::from_iter(guard.iter().filter(|app| app.app.uuid == app_id).next().expect("Cannot find the specified app").output.iter().map(|sender| sender.clone()))
        };
        for sender in senders {
            for message in all_message.clone() {
                sender.send(message).await.expect("Cannot write data");
            }
        }
        Result::Ok(Response::new(PublishPipeMessageResponse {
            result: Some(GenericResult {
                message: "Publish succeed".to_string(),
                succeed: true,
            })
        }))
    }

    async fn list_pipes_and_apps(&self, request: Request<Empty>) -> Result<Response<ListPipesAndAppsResponse>, Status> {
        let apps = {
            Vec::from_iter(self.apps.lock().expect("Cannot acquire the apps lock").iter().map(|item| item.app.clone()))
        };
        let pipes = {
            Vec::from_iter(self.pipes.lock().expect("Cannot acquire pipe lock").iter().map(|item| protocol_buffer::entity::MessagePipe {
                uuid: item.uuid.clone(),
                name: item.name.clone(),
                description: item.description.clone(),
                has_table: item.has_table,
                storing_importance: match item.message_importance {
                    StoringImportance::Critical => 1,
                    StoringImportance::Usual => 0,
                }
            }))
        };
        let message = ListPipesAndAppsResponse {
            pipes: pipes,
            apps: apps,
            result: Some(GenericResult {
                message: "Succeed".into(),
                succeed: true,
            })
        };
        Ok(Response::new(message))
    }
}



